package ba.keco.hbase.rowkey;

import java.util.UUID;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import ba.keco.hbase.lib.HBaseShowcaseInputFormat;
import ba.keco.hbase.lib.Util;

public class MapReduceTest extends Configured implements Tool{

	public static void main(String[] args) throws Exception {
		ToolRunner.run(new MapReduceTest(), args);
	}
	
	@Override
	public int run(String[] arg0) throws Exception {
		int cfs = getConf().getInt("hbase.showcase.cf.number", -1);
		boolean isRandom = getConf().getBoolean("hbase.showcase.rowkey.israndom", true);
		boolean keepTable = getConf().getBoolean("hbase.showcase.keep.table", true);

		String tableName = getConf().get("hbase.showcase.table.name");

		try {
			Util.delete(tableName);
		} catch (Exception e) {
		}
		Util.createTable(tableName, cfs);

		getConf().set(TableOutputFormat.OUTPUT_TABLE, tableName);

		Job job = new Job(getConf(), "RowKey-test-" + isRandom);

		job.setJarByClass(MapReduceTest.class);
		
		job.setNumReduceTasks(0);
		job.setInputFormatClass(HBaseShowcaseInputFormat.class);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.setMapperClass(RowKeyTestMapper.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Put.class);

		long startTime = System.currentTimeMillis();
		boolean isSuccessful = job.waitForCompletion(true);
		long endTime = System.currentTimeMillis();
		System.out.println("###," + cfs + "," + isRandom + "," + (endTime - startTime));
		if (!keepTable) {
			Util.delete(tableName);
		}
		return isSuccessful ? 0 : 1;
	}
	
	public static class RowKeyTestMapper extends Mapper<NullWritable, NullWritable, NullWritable, Put>{
		
		private boolean isRandom = false;
		
		private int mapperId; 
		
		private int key = 0;
		
		private int cfs;
		
		private int qfs;
		
		protected void setup(Context context) throws java.io.IOException ,InterruptedException {
			mapperId = context.getTaskAttemptID().getTaskID().getId() + 1;
			isRandom = context.getConfiguration().getBoolean("hbase.showcase.rowkey.israndom", true);
			cfs = context.getConfiguration().getInt("hbase.showcase.cf.number", -1);
			qfs = context.getConfiguration().getInt("hbase.showcase.qf.number", -1);
		}
		
		protected void map(NullWritable key, NullWritable value, Context context) throws java.io.IOException ,InterruptedException {
			String rowkey = rowkeyGenerator(isRandom);
			Put put = Util.generateRecord(rowkey, cfs, qfs);
			context.write(NullWritable.get(), put);
		}
		
		private String rowkeyGenerator(boolean isRandom){
			if (isRandom){
				return UUID.randomUUID().toString();
			}else{
				key = key + mapperId;
				return String.format("%10d", key);
			}
		}
	}

}
